package cn.wisenergy.pai.monitor.collector;

import java.io.IOException;
import java.util.Set;

import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.message.MessageAndMetadata;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.wisenergy.pai.monitor.linux.parser.TimeNode;

public class HBaseDataProcessor implements TimeNodeProcessor {
	private static final Logger LOG = LoggerFactory
			.getLogger(HBaseDataProcessor.class);
	private KafkaStream<byte[], byte[]> stream;
	private int threadNumber;

	@Override
	public void run() {
		ConsumerIterator<byte[], byte[]> it = stream.iterator();
		while (it.hasNext()) {
			try {
				MessageAndMetadata<byte[], byte[]> mm = it.next();
				String key = new String(mm.key());
				String[] splited = key.split("_");
				if (splited.length == 2) {
					String ip = splited[0];
					long time = Long.parseLong(splited[1]);
					TimeNode tn = new TimeNode();
					try {
						tn.toObject(mm.message());
					} catch (IOException e) {
						LOG.warn("can't parse the timenode for "+key);
						continue;
					}
				}
			} catch (Exception e) {
				LOG.warn("can't parse the timenode anymore");
			}
		}
	}

	@Override
	public void init(KafkaStream<byte[], byte[]> str, int num) {
		stream = str;
		threadNumber = num;
	}

}
